package cn.itcast.model.base

import java.util.Properties

import cn.itcast.model.bean.HBaseMeta
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

trait BaseModel {
  def setAppName(): String

  def importSpark()

  def setTagId(): Int

  def computeTag(fiveRuleDF: DataFrame, hbaseSource: DataFrame): DataFrame

  //1. 获取SparkSession对象
  val spark: SparkSession = SparkSession.builder()
    .appName(setAppName)
    .master("local[*]")
    .getOrCreate()

  import spark.implicits._
  import org.apache.spark.sql.functions._

  def loadMysqlSource() = {
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val table = "tbl_basic_tag"
    val properties = new Properties()
    val mysqlSource: DataFrame = spark.read.jdbc(url, table, properties)
    mysqlSource
  }

  def getFourRuleData(mysqlSource: DataFrame) = {
    mysqlSource.select($"rule")
      .where("id = " + setTagId())
      .rdd.map(row => {
      val ruleStr: String = row.getAs[String]("rule")
      val map: Map[String, String] = ruleStr.split("##")
        .map(line => {
          val arr: Array[String] = line.split("=")
          (arr(0), arr(1))
        }).toMap
      map
    }).collect()(0)
  }

  def getFiveRuleData(mysqlSource: DataFrame) = {
    mysqlSource.select($"id", $"rule")
      .where("pid = " + setTagId())
  }
  def loadHBaseSource(param: Map[String, String]): DataFrame = {
    spark.read
      .format("cn.itcast.model.util.HbaseSource")
      .options(param)
      .load()
  }
  def loadOldDF(): DataFrame = {
    spark.read
      .format("cn.itcast.model.util.HbaseSource")
      .option(HBaseMeta.ZKHOSTS, "192.168.10.20")
      .option(HBaseMeta.ZKPORT, "2181")
      .option(HBaseMeta.HBASETABLE, "test34") //加载用户画像表
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .load()
  }

  def mergeTag(newDF: DataFrame, oldDF: DataFrame) = {
    val joinDF: DataFrame = newDF.join(oldDF, newDF.col("userId") === oldDF.col("userId"),
      "full")
    val mergeTag: UserDefinedFunction = udf((newTag: String, oldTag: String) => {
      if (StringUtils.isBlank(newTag)) {
        oldTag
      } else if (StringUtils.isBlank(oldTag)) {
        newTag
      } else if (StringUtils.isNotBlank(newTag) && StringUtils.isNotBlank(oldTag)) {
        val tmpStr: String = newTag + "," + oldTag
        val set: Set[String] = tmpStr.split(",").toSet
        val mergeTagStr: String = set.mkString(",")
        mergeTagStr
      } else {
        ""
      }
    })
    val result: DataFrame = joinDF.select(
      when(newDF.col("userId").isNotNull, oldDF.col("userId"))
        .when(newDF.col("userId").isNull, oldDF.col("userId"))
        .as("userId")
      ,
      mergeTag(newDF.col("tagIds"), oldDF.col("tagIds")).as("tagIds")
    )
    result
  }
  def saveResult(result: DataFrame) = {
    result.write
      //指定数据源
      .format("cn.itcast.model.util.HbaseSource")
      .option(HBaseMeta.ZKHOSTS, "192.168.10.20")
      .option(HBaseMeta.ZKPORT, "2181")
      .option(HBaseMeta.HBASETABLE, "test34") //加载用户画像表
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .save()
  }

  def executeModel()={
    val mysqlSource: DataFrame = loadMysqlSource()
    val fourRuleMap: Map[String, String] = getFourRuleData(mysqlSource)
    val fiveRuleDF: Dataset[Row] = getFiveRuleData(mysqlSource)
    val hbaseSource: DataFrame = loadHBaseSource(fourRuleMap)
    val newDF: DataFrame = computeTag(fiveRuleDF,hbaseSource)
    val oldDF: DataFrame = loadOldDF()
    val result: DataFrame = mergeTag(newDF,oldDF)
    saveResult(result)
  }
}
